Cheatsheets

Appearance

LangGraph Checkpoint 表关系 ​

本文档说明 LangGraph Checkpoint 持久化机制与应用会话（Conversation）之间的关系。

概述 ​

LangGraph 使用 MySQL Checkpointer 来持久化对话状态，实现多轮对话的自动记忆。这涉及以下几个数据库表：

  • checkpoints - 检查点快照（链式结构）
  • checkpoint_writes - 写入记录（增量数据）
  • checkpoint_blobs - 大对象存储（channel 历史版本）

表结构详解 ​

1. checkpoints 表 ​

检查点快照表，存储对话状态的完整快照，形成链式历史结构。

字段类型说明
thread_idvarchar(150)会话线程 ID，格式为 conv_{conversation_id}
checkpoint_nsvarchar(2000)命名空间
checkpoint_idvarchar(150)当前检查点 ID（UUID 格式）
parent_checkpoint_idvarchar(150)父检查点 ID（形成链式结构）
typevarchar(150)检查点类型
checkpointjson检查点数据（JSON 格式）
metadatajson元数据
checkpoint_ns_hashbinary(16)命名空间哈希

链式结构示意：

checkpoint_1 (parent: None)
    ↓
checkpoint_2 (parent: checkpoint_1)
    ↓
checkpoint_3 (parent: checkpoint_2)
    ↓
...

2. checkpoint_writes 表 ​

写入记录表，存储每个检查点的增量写入数据。

字段类型说明
thread_idvarchar(150)会话线程 ID
checkpoint_nsvarchar(2000)命名空间
checkpoint_idvarchar(150)关联到 checkpoints 表
task_idvarchar(150)任务 ID（每次 LLM 调用生成一个）
idxint索引
channelvarchar(150)通道名（如 messages, branch:to:agent）
typevarchar(150)数据类型（如 msgpack, null）
bloblongblob实际数据（二进制）
checkpoint_ns_hashbinary(16)命名空间哈希
task_pathvarchar(2000)任务路径

常见 channel 类型：

  • messages - 对话消息
  • branch:to:agent - 分支路由信息
  • __start__ - 起始状态

3. checkpoint_blobs 表 ​

大对象存储表，存储 channel 的历史版本数据。

字段类型说明
thread_idvarchar(150)会话线程 ID
checkpoint_nsvarchar(2000)命名空间
channelvarchar(150)通道名
versionvarchar(150)版本号
typevarchar(150)数据类型
bloblongblob实际数据
checkpoint_ns_hashbinary(16)命名空间哈希

应用会话与 Checkpoint 的关系 ​

会话 ID 映射 ​

应用中的 conversation_id 与 LangGraph 的 thread_id 映射关系：

thread_id = f"conv_{conversation_id}"

示例：

  • 应用会话 ID: 0aec2add-0d09-4777-9c71-7a1f695294b5
  • LangGraph thread_id: conv_0aec2add-0d09-4777-9c71-7a1f695294b5

架构图 ​

┌─────────────────────────────────────────────────────────────────────────┐
│                           应用层                                         │
├─────────────────────────────────────────────────────────────────────────┤
│  routes/chat.py                                                          │
│  ├── /api/chat          → chat_service.chat()                           │
│  └── /api/chat/stream   → chat_service.chat_stream()                    │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                        服务层                                            │
├─────────────────────────────────────────────────────────────────────────┤
│  services/langgraph_chat.py                                              │
│  └── LangGraphChatService                                                │
│      ├── chat()         → 普通对话                                       │
│      └── chat_stream()  → 流式对话                                       │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                       LangGraph 层                                       │
├─────────────────────────────────────────────────────────────────────────┤
│  StateGraph(MessagesState) + Checkpointer                                │
│  └── 自动管理对话状态持久化                                               │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                       数据库层                                           │
├─────────────────────────────────────────────────────────────────────────┤
│  MySQL Checkpoint 表                                                     │
│  ├── checkpoints         → 检查点快照链                                   │
│  ├── checkpoint_writes   → 增量写入记录                                   │
│  └── checkpoint_blobs    → 大对象存储                                     │
└─────────────────────────────────────────────────────────────────────────┘

数据流转 ​

  1. 用户发送消息：

    • 前端调用 /api/chat/stream
    • 使用 conversation_id 生成 thread_id

  2. LangGraph 处理：

    • 根据 thread_id 从 checkpoint 恢复历史消息
    • 调用 LLM 生成回复
    • 将新消息追加到 checkpoint

  3. 持久化：

    • 创建新的 checkpoint 记录（parent_checkpoint_id 指向上一个）
    • 写入 checkpoint_writes 记录增量数据
    • 更新 checkpoint_blobs 存储 channel 数据

常用查询 ​

查询特定会话的 checkpoint 历史 ​

sql
-- 查询会话的所有检查点
SELECT
    checkpoint_id,
    parent_checkpoint_id,
    JSON_EXTRACT(checkpoint, '$.ts') as timestamp
FROM checkpoints
WHERE thread_id = 'conv_{conversation_id}'
ORDER BY checkpoint_id;

查询特定会话的写入记录 ​

sql
-- 查询会话的所有写入记录
SELECT
    checkpoint_id,
    task_id,
    channel,
    type,
    LENGTH(`blob`) as blob_size
FROM checkpoint_writes
WHERE thread_id = 'conv_{conversation_id}'
ORDER BY checkpoint_id;

查询会话的消息历史 ​

sql
-- 查询 messages channel 的数据
SELECT
    channel,
    version,
    type,
    LENGTH(`blob`) as blob_size
FROM checkpoint_blobs
WHERE thread_id = 'conv_{conversation_id}'
  AND channel = 'messages'
ORDER BY version;

数据迁移 ​

从旧的 messages 表迁移到 LangGraph checkpoint 的策略：

  1. 首次访问迁移：当用户访问旧对话时，自动将历史消息迁移到 checkpoint
  2. 增量迁移：新对话直接使用 checkpoint，旧对话按需迁移
  3. 兼容性：保留原有的 conversations 和 messages 表用于前端展示

迁移代码位于 services/migration.py。

相关代码文件 ​

文件说明
services/checkpointer.pyMySQL Checkpointer 初始化
services/langgraph_chat.pyLangGraph 聊天服务核心
services/migration.py旧对话迁移工具
routes/chat.py聊天路由（调用 LangGraph 服务）

参考资料 ​